package com.zoe.flume.source.custom;

import com.google.common.collect.Lists;
import org.apache.flume.*;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.source.AbstractSource;

import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by shenyuhang on 2018/2/1.
 */
public class CustomSource extends AbstractSource implements
        PollableSource, Configurable {
    private AtomicInteger id;
    private int batchSize = 100;

    @Override
    public Status process() throws EventDeliveryException {
        Status status = Status.READY;
        try {
            TimeUnit.MILLISECONDS.sleep(5);
            List<Event> events = Lists.newArrayListWithExpectedSize(batchSize);

            while (id.get() <= batchSize) {
                if (!id.compareAndSet(batchSize, 0)) {
                    String body = String.format("{id: '%s', content: 'hello elasticsearch%s'}", id.get(), id.get());
                    Event event = EventBuilder.withBody(body.getBytes());
                    events.add(event);
                    id.incrementAndGet();
                } else {
                    getChannelProcessor().processEventBatch(events);
                    events.parallelStream().forEach(event-> event = null);
                    events.clear();
                    events = Lists.newArrayListWithExpectedSize(batchSize);
                    Thread.sleep(100000);
                }
            }

        } catch (Throwable t) {
            status = Status.BACKOFF;
        }
        return status;
    }

    @Override
    public synchronized void start() {
        super.start();
        id = new AtomicInteger(0);
    }

    @Override
    public long getBackOffSleepIncrement() {
        return 0;
    }

    @Override
    public long getMaxBackOffSleepInterval() {
        return 0;
    }

    @Override
    public void configure(Context context) {

    }
}
